feat(sdks): close Node and Java per-task policy and batching parity#436
Merged
Conversation
The dispatcher spawned every job it was handed and immediately took the next, so the job channel drained as fast as the scheduler filled it. A channel that never fills applies no backpressure, and this SDK never set max_in_flight — which Python and Java both do — so nothing capped in-flight work. The worker claimed jobs it could not run, stranding them Running while peers sharing the database skipped them. Add a `concurrency` option, separate from `channelCapacity`, mirroring Java: it caps what the scheduler claims and what the dispatcher runs. The dispatcher permit is the only bound on the mesh path, where stolen jobs never reach this scheduler's in-flight accounting. Opt-in — unset stays unbounded, so no existing worker silently acquires a cap. channelCapacity's doc claimed to be the in-flight bound and never was; it now says what it does.
Python has had this since the per-task bulkhead landed; this SDK hardcoded None, so one slow task could occupy a worker's whole pool. The config collector gated on a hand-listed set of option names, which drops any option missing from the list — silently, and invisibly to type-checking. Derive the gate from the built config instead, so a new option cannot be lost the next time. Name the field in the rate-spec error too, now that more than one option shares that grammar.
Python and Node have had both for a while; this SDK carried neither. The hardcoded Nones were not a lazy binding — the wire struct had no field for them, so nothing upstream could send one. Available on Task and on @TaskHandler, matching how the retry curve and circuit breaker already reach the worker. A malformed rate spec now fails the worker's start, naming the task and option, rather than running the task unthrottled; registration gained an error path to allow that. Worker holds the tasks that carry policy rather than a map per knob, so the encoder reads one source: the next knob is encoded in one place instead of needing its own map, its own capture, and its own arm of a name union.
The drain loop handled one result per wake, so each job cost its own transaction. Take everything already queued and finalize it in one, matching Python. The channel is bounded, so the drain caps itself. Successes batch into a single write while failures stay per-result, and one outcome still comes back per job in order, so events and middleware fire exactly once each. The catch_unwind now guards a batch, widening a panic from one dropped outcome to N — acceptable because the realistic failure is a batch write error, which already falls back to finalizing job by job.
The drain loop handled one result per wake, so each job cost its own transaction. Take everything already queued and finalize it in one, matching Python and Node. The batch is capped explicitly rather than drained until empty: unlike the other shells, this result channel is unbounded so the dispatcher never blocks a runtime worker, which means nothing else would bound one drain. The JNI local frame stays per outcome — sized for a single callback, it would overflow around a batch.
Caps how fast a task may retry across all of its jobs; once spent, failures dead-letter rather than retrying. Landed for Python in #435. Nothing was added to the config gate for it — the gate derives from the built config, so a new option reaches the scheduler on its own. The rate parser now names the field, since two options share its "100/m" grammar.
Caps how fast a task may retry across all of its jobs; once spent, failures dead-letter rather than retrying. Landed for Python in #435. On Task and @TaskHandler beside the rate limit, sharing its parser and its fail-fast on a malformed spec. Completes this SDK's per-task policy surface.
The last of the four per-task knobs this SDK carried no wire field for. All three SDKs now surface every one, so no binding hardcodes a None the caller cannot reach.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds end-to-end task policy configuration for rate limits, retry budgets, concurrency caps, and per-task in-flight limits across Java and Node APIs. Worker dispatch is concurrency-bounded, results are processed in batches, and validation covers enforcement and startup errors. ChangesTask policy runtime
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Worker
participant Dispatcher
participant Scheduler
participant Callback
Client->>Worker: configure task policies and concurrency
Worker->>Dispatcher: start bounded execution
Dispatcher->>Scheduler: submit completed result batch
Scheduler->>Worker: return ordered outcomes
Worker->>Callback: report each outcome
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (3)
sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java (1)
23-58: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExercise
maxConcurrentacross two workers.This single-worker test also passes if the policy is mistakenly implemented as worker-local
maxInFlightPerTask. Start two workers against the same database and assert their shared peak remains one to validate the cluster-wide contract.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java` around lines 23 - 58, Extend maxConcurrentSerializesOneTask to start two workers against the same database and enqueue enough solo jobs to exercise both workers, while retaining the running/peak tracking across both handlers. Await completion for every job, then assert the shared peak remains one, ensuring maxConcurrent is enforced cluster-wide rather than per worker.sdks/node/test/worker/backpressure.test.ts (1)
29-64: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftTest the dispatcher limit independently from scheduler claim bounding.
This test still passes without the semaphore because
SchedulerConfig.max_in_flightis also set toconcurrency. Add a direct dispatcher test or a mesh-enabled case where stolen jobs bypass this scheduler’s accounting.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/node/test/worker/backpressure.test.ts` around lines 29 - 64, Update the concurrency test around queue.runWorker so it validates the dispatcher’s own in-flight limit independently of SchedulerConfig.max_in_flight. Add a direct dispatcher-focused setup or enable a mesh/stolen-job path that bypasses scheduler accounting, then retain assertions that peak running jobs stays at or below concurrency and exceeds one.sdks/java/src/main/java/org/byteveda/taskito/task/Task.java (1)
52-258: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftRepeated 11-arg constructor call across every wither method.
Every copy-site (
withOptions,retryPolicy,codecs,idempotent,circuitBreaker,rateLimit,maxConcurrent,retryBudget,maxInFlightPerTask, plus bothof(...)factories) hand-duplicates the same 11-argumentnew Task<>(...)call. All sites are correctly wired today, but this is exactly the failure mode the PR fixes on the Node side (buildTaskConfigs/setsSomethingderiving from a single built config so a new option can't be silently dropped) — here, adding the next policy field means touching 11 call sites by hand, and a missed one fails silently (wrong field slot, no compile error since all fields share compatible types in places).Consider a private "copy" helper (e.g. a mutable holder or an internal builder) that every fluent method delegates to, so a new field is added in one place.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/java/src/main/java/org/byteveda/taskito/task/Task.java` around lines 52 - 258, Centralize Task copy construction to eliminate the repeated 11-argument new Task calls across the of factories and fluent methods including withOptions, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent, retryBudget, and maxInFlightPerTask. Add a private copy/helper or internal builder that preserves all existing fields and lets each method override only its targeted value, then delegate every factory and wither through it so future fields are wired in one place.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/taskito-java/src/worker.rs`:
- Line 109: Move the fallible register_task_policies call before
register_live_worker and register_subscriptions in the worker initialization
flow, so malformed rate specifications are rejected before persistent worker
state or subscriptions are created. Preserve the existing policy registration
behavior and successful initialization order after validation.
In `@crates/taskito-node/src/convert/task_config.rs`:
- Around line 18-25: Update parse_rate_spec and its callers to accept the
configured task name from input.name, then include that task name alongside the
option field in malformed-policy errors. Apply the same task-aware error context
to the related parsing path noted in the comment, while preserving the existing
validation and fail-fast behavior.
- Line 48: Update the max_in_flight_per_task conversion to map 0 to None for
uncapped behavior, preserve positive values as usize, and reject negative values
rather than clamping them to 1. Change only the conversion logic for
max_in_flight_per_task.
In `@crates/taskito-node/src/worker.rs`:
- Around line 201-208: Limit the result collection in the worker loop around
result_rx.recv() and result_rx.try_recv() to at most the configured channel
capacity. Stop draining once the batch reaches that capacity, while preserving
the existing batched finalization behavior for smaller batches.
In `@sdks/java/src/main/java/org/byteveda/taskito/task/Task.java`:
- Around line 202-215: The maxConcurrent method must reject zero as a
concurrency cap while preserving null as the uncapped value. Update
Task.maxConcurrent to normalize a zero input to null before constructing the new
Task, leaving positive limits unchanged.
In `@sdks/node/test/worker/resultBatching.test.ts`:
- Around line 61-64: Extend the result-batching assertions around completed and
dead job IDs to verify the two collections are disjoint, matching the Java
test’s overlap check. Preserve the existing length and per-collection uniqueness
assertions, and add an assertion that no ID appears in both completed and dead.
- Around line 52-57: Update the test setup around the enqueue loop and
queue.runWorker call so every asynchronous queue.enqueue operation is awaited
before the worker starts. Collect or await both enqueues for each iteration,
then invoke runWorker only after the complete burst is queued, preserving the
existing concurrency and batchSize options.
---
Nitpick comments:
In `@sdks/java/src/main/java/org/byteveda/taskito/task/Task.java`:
- Around line 52-258: Centralize Task copy construction to eliminate the
repeated 11-argument new Task calls across the of factories and fluent methods
including withOptions, retryPolicy, codecs, idempotent, circuitBreaker,
rateLimit, maxConcurrent, retryBudget, and maxInFlightPerTask. Add a private
copy/helper or internal builder that preserves all existing fields and lets each
method override only its targeted value, then delegate every factory and wither
through it so future fields are wired in one place.
In `@sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java`:
- Around line 23-58: Extend maxConcurrentSerializesOneTask to start two workers
against the same database and enqueue enough solo jobs to exercise both workers,
while retaining the running/peak tracking across both handlers. Await completion
for every job, then assert the shared peak remains one, ensuring maxConcurrent
is enforced cluster-wide rather than per worker.
In `@sdks/node/test/worker/backpressure.test.ts`:
- Around line 29-64: Update the concurrency test around queue.runWorker so it
validates the dispatcher’s own in-flight limit independently of
SchedulerConfig.max_in_flight. Add a direct dispatcher-focused setup or enable a
mesh/stolen-job path that bypasses scheduler accounting, then retain assertions
that peak running jobs stays at or below concurrency and exceeds one.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5b8c29d6-c1dd-42aa-83f3-41437e1b6cc9
📒 Files selected for processing (17)
crates/taskito-java/src/convert.rscrates/taskito-java/src/worker.rscrates/taskito-node/src/config.rscrates/taskito-node/src/convert/task_config.rscrates/taskito-node/src/dispatcher.rscrates/taskito-node/src/worker.rssdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.javasdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.javasdks/java/src/main/java/org/byteveda/taskito/task/Task.javasdks/java/src/main/java/org/byteveda/taskito/worker/Worker.javasdks/java/src/test/java/org/byteveda/taskito/core/ResultBatchingTest.javasdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.javasdks/node/src/types.tssdks/node/src/worker.tssdks/node/test/worker/backpressure.test.tssdks/node/test/worker/resultBatching.test.tssdks/node/test/worker/taskConfig.test.ts
A malformed rate spec was rejected after the worker row and its subscriptions were already written, and the start then returned without a handle — so the lifecycle loop that removes them never ran and both rows leaked. Build the policies first: parsing is pure, so a rejected config now leaves nothing behind.
A bounded channel caps how many results sit in it at once, not how many one drain takes: senders refill slots while try_recv runs, so the loop could swallow a whole backlog into one Vec and stall finalization behind it. Cap it like the Java drain does.
A malformed spec on Node named the option but not the task, so a start-up failure could not say which config was wrong. Java already named both. Task.maxConcurrent(0)/maxInFlightPerTask(0) stored the zero verbatim, and the scheduler reads a zero cap as "never dispatch". Zero is the annotation's unset sentinel, so the fluent setters now read it the same way. Also awaits the enqueues in the batching test — they could still be pending when the worker started — and asserts the terminal ids are disjoint, which per-list uniqueness would miss.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/taskito-node/src/convert/task_config.rs`:
- Around line 19-31: Update parse_rate_spec to accept a scope/resource label
alongside field and name, and use it in the invalid_arg message instead of
hardcoding “task”; update both task and queue call sites, including the usage
around line 94, to pass the appropriate “task” or “queue” scope so malformed
queue limits identify the queue correctly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 94e63b61-a58c-4d72-aa10-7b751341520f
📒 Files selected for processing (5)
crates/taskito-java/src/worker.rscrates/taskito-node/src/convert/task_config.rscrates/taskito-node/src/worker.rssdks/java/src/main/java/org/byteveda/taskito/task/Task.javasdks/node/test/worker/resultBatching.test.ts
🚧 Files skipped from review as they are similar to previous changes (4)
- sdks/node/test/worker/resultBatching.test.ts
- crates/taskito-node/src/worker.rs
- sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
- crates/taskito-java/src/worker.rs
Queues configure a rate limit through the same parser, so wording the error "on task" reported a malformed queue limit as a task — a regression from adding the name in the first place. Pass the kind alongside the name.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@sdks/node/test/worker/rateScope.test.ts`:
- Line 14: Strengthen the assertions around queue.runWorker in the rate-scope
tests so they verify the complete error scope, not merely a substring. Capture
the thrown error message once for the task-scoped and queue-scoped cases, then
assert the expected task or queue scope is present and the opposite scope is
absent, preserving the distinction required by the contract.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: eab5c3a7-0163-4115-ae68-269c959d15d0
📒 Files selected for processing (2)
crates/taskito-node/src/convert/task_config.rssdks/node/test/worker/rateScope.test.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- crates/taskito-node/src/convert/task_config.rs
toThrow(string) only checks containment, so a message naming both scopes would have satisfied it — and naming exactly one is the contract under test. Compare the whole message.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes the Node/Java half of the parity debt each scaling tier left behind. One live bug, the rest missing surface.
The bug
NodeDispatcher::runspawned every job it was handed and immediately took the next, and this SDK never setmax_in_flight— Python and Java both do. A channel that never fills applies no backpressure, so nothing capped in-flight work: the worker claimed jobs it could not run, stranding themRunningwhile peers sharing the database skipped them. Same over-claim class as S01.types.tsalready documented a bound that did not exist.Fixed with a
concurrencyoption, separate fromchannelCapacityand mirroring Java. It bounds what the scheduler claims and what the dispatcher runs — the dispatcher permit matters on its own because it is the only bound on the mesh path, where stolen jobs never reach this scheduler's in-flight accounting. Opt-in: unset stays unbounded, so no existing worker silently acquires a cap.channelCapacity's doc now says what it does.Per-task policy — all four knobs, all three SDKs
Java surfaced none of the four. The hardcoded
Nones were not a lazy binding: the wire struct had no field, so nothing upstream could send a value.rateLimitmaxConcurrentretryBudgetmaxInFlightPerTaskNo binding hardcodes a
Nonethe caller cannot reach any more. On Java these live onTaskand@TaskHandler, beside the retry curve and circuit breaker that already reach the worker that way. A malformed rate spec now fails the worker's start, naming the task and the option, rather than running it unthrottled — registration gained an error path for that.Two "hand-list every field" traps removed, not extended
Workernow holds the tasks that carry policy, so the encoder reads one source.Both paid off inside this PR: when
retryBudgetunblocked, Node needed no gate edit and Java needed oneifin one encoder.Result batching
Both SDKs handled one result per wake, so every job cost its own transaction. They now finalize everything already queued in one, matching Python.
Java caps the batch explicitly. Its result channel is deliberately unbounded so the dispatcher never blocks a runtime worker, which means Python's "the channel is bounded, so the drain caps itself" reasoning does not transfer — inheriting it uncritically was the trap. Node's channel is bounded, so it needs no cap. Java's JNI local frame stays per outcome; sized for one callback, it would overflow around a batch.
Verification
Every fix has a test watched failing with the fix removed:
expected 12 to be less than or equal to 3expected 4 to be 1retryBudgetThe Node backpressure test sets
batchSizeto the whole backlog on purpose: several bounds overlap (poll cadence, task duration, channel capacity, in-flight cap) and the test proves nothing unless the one under test is the tightest. An earlier version passed without the fix because the ~50ms poll cadence was the real limiter.Node 404 tests, typecheck + biome clean. Java
./gradlew build(tests + spotless + checkstyle) plusplainJavadocJar, which gates publishing rather thanbuild. Rust 190 tests across default/postgres/redis/native-async, clippy + fmt clean.Follow-ups, deliberately not here
Task's constructor went 7 → 11 positional args following the existing wither idiom. The next knob makes it 12; it wants grouping into aTaskPolicyfirst. Separate concern from surfacing options.Summary by CodeRabbit
rateLimit,retryBudget,maxConcurrent, andmaxInFlightPerTask(Java SDK, Node worker, and native task policy).rateLimit/policy specs now fail worker startup early with scope-aware validation errors.